set hive.exec.dynamic.partition = true;
set hive.exec.dynamic.partition.mode = 'nonstrict';
set hive.exec.max.dynamic.partitions = 400;
set hive.exec.max.dynamic.partitions.pernode = 400;
with waybill_sign as (
    select
        pick_network_code
         ,pick_network_name
         ,dispatch_network_code
         ,dispatch_network_name
         ,substring(COALESCE(taking_early_scantime,taking_later_scantime),12,8) as taking_time
         ,datediff(substring(COALESCE(instore_later_scantime,outstore_later_scantime,sign_later_scantime),1,10),substring(COALESCE(taking_early_scantime,taking_later_scantime),1,10)) sign_days
         ,count(waybill_no) as waybill_count
    from jms_dwd.dwd_wide_sign_summary_waybill_dt
    where  dt >=date_sub('{{ execution_date | cst_ds }}',6) and dt<='{{ execution_date | cst_ds }}' and COALESCE(taking_early_scantime,taking_later_scantime) is not null
    group by
        pick_network_code
           ,pick_network_name
           ,dispatch_network_code
           ,dispatch_network_name
           ,substring(COALESCE(taking_early_scantime,taking_later_scantime),12,8)
           ,datediff(substring(COALESCE(instore_later_scantime,outstore_later_scantime,sign_later_scantime),1,10),substring(COALESCE(taking_early_scantime,taking_later_scantime),1,10))
),
in_warehouse as (
    select
        network_code,
        latest_arrival_time,
        warehouse_start_time,
        warehouse_end_time,
        receipt_end_time,
        if(rnk =1,concat_ws ('-','00:00',receipt_end_time),latest_warehousing_time_quantum) as latest_warehousing_time_quantum
    from
        (
            select network_code,
                   latest_arrival_time,
                   warehouse_start_time,
                   warehouse_end_time,
                   receipt_end_time,
                   receipt_start_time,
                   concat_ws ('-',receipt_start_time,receipt_end_time) as latest_warehousing_time_quantum,
                   row_number() over(partition by network_code order by receipt_start_time) as rnk
            from jms_dim.dim_yl_tmsnew_tms_in_warehouse_shi_base_dt
            where STATUS = 1 and IS_DELETE = 1 and IS_ENABLE=1 and dt = '{{ execution_date | cst_ds }}'
        ) a
    union all
    select
        network_code,
        latest_arrival_time,
        warehouse_start_time,
        warehouse_end_time,
        '23:59' as receipt_end_time,
        concat_ws ('-',from_unixtime(unix_timestamp(concat('2099-01-01 ',receipt_end_time,':00'))+60,'HH:mm'),'23:59') as latest_warehousing_time_quantum
    from
        (
            select a.network_code,
                   a.latest_arrival_time,
                   a.warehouse_start_time,
                   a.warehouse_end_time,
                   a.receipt_end_time,
                   a.receipt_start_time,
                   concat_ws ('-',a.receipt_start_time,a.receipt_end_time) as latest_warehousing_time_quantum,
                   row_number() over(partition by a.network_code order by a.receipt_end_time desc ) as rnk
            from jms_dim.dim_yl_tmsnew_tms_in_warehouse_shi_base_dt a
                left anti join (select network_code from jms_dim.dim_yl_tmsnew_tms_in_warehouse_shi_base_dt where where STATUS = 1 and IS_DELETE = 1 and IS_ENABLE=1 and dt = ' {{ execution_date | cst_ds }} ' and receipt_end_time='23:59') b
            on a.network_code = b.network_code
            where a.STATUS = 1 and a.IS_DELETE = 1 and a.IS_ENABLE=1 and a.dt = '{{ execution_date | cst_ds }}'
        ) a
    where rnk=1
)
insert overwrite table jms_tmp.tmp_waybill_sign_taking_time_quantum_dt partition(dt)
select
    pick_network_code
     ,pick_network_name
     ,dispatch_network_code
     ,dispatch_network_name
     ,latest_warehousing_time_quantum
     ,sign_days
     ,waybill_count
     ,''
     ,''
     ,''
     ,''
     ,'{{ execution_date | cst_ds }}'
from
    (
        select
            pick_network_code
             ,pick_network_name
             ,dispatch_network_code
             ,dispatch_network_name
             ,latest_warehousing_time_quantum
             ,sign_days
             ,sum(waybill_count) as waybill_count
             ,row_number() over(partition by pick_network_code,dispatch_network_code,latest_warehousing_time_quantum order by sum(waybill_count) desc) rn
        from waybill_sign
                 left join in_warehouse
                           on waybill_sign.pick_network_code = in_warehouse.network_code and taking_time>=concat(split(latest_warehousing_time_quantum,'-')[0],':00') and taking_time<=concat(split(latest_warehousing_time_quantum,'-')[1],':59')
        group by
            pick_network_code
               ,pick_network_name
               ,dispatch_network_code
               ,dispatch_network_name
               ,latest_warehousing_time_quantum
               ,sign_days
    )a where rn = 1